package org.finesys.message.consumer;

import com.fasterxml.jackson.databind.JsonNode;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.finesys.commom.jackson.util.JacksonUtil;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

/**
 * 监听mq消息的监听器
 */
@Component("mqMessageListener")
@Slf4j
@RequiredArgsConstructor
public class MqMessageListener {


    // 是否自动提交偏移量
    @Value("${spring.cloud.stream.bindings.consumer.enableAutoCommit}")
    private boolean enableAutoCommit;

    /**
     * 创建一个名为"kafkaInput"的Kafka消费者Bean。
     * 函数名称 kafkaInput 对应于配置文件的spring.cloud.stream.bindings.kafkaInput-in-0，此处为"kafkaInput-in-0"。
     *
     * @return Kafka消费者，消费字符串类型的消息。
     */
    @Bean("kafkaInput")
    public Consumer<String> kafkaInput() {
        System.out.println("初始化kafka监听器");
        return this::processMessage;
    }

    /**
     * 处理接收到的消息。
     *
     * @param message 接收到的消息内容
     */
    private void processMessage(String message) {
        System.out.println("收到消息：" + message);
        if (log.isDebugEnabled()) {
            log.debug("接收到消息：{}", message);
        }
        //手工确认偏移量
        if (!enableAutoCommit) {
            // 创建匿名内部类实现Acknowledgment接口
            Acknowledgment acknowledgment = () -> {
                // 手工确认偏移量逻辑
                System.out.println("手工确认消息：" + message);
            };
            //处理消息逻辑并手工确认偏移量
            handleAcknowledgment(message, acknowledgment);
        } else {
            doMessage(message);
        }
    }


    /**
     * 处理消息并手工确认偏移量。
     *
     * @param message        接收到的消息内容
     * @param acknowledgment 用于确认消息消费的偏移量
     */
    private void handleAcknowledgment(String message, Acknowledgment acknowledgment) {
        System.out.println("手工确认消息：" + message);
        if (log.isInfoEnabled()) {
            log.info("手工确认消息：{}", message);
        }
        //处理消息逻辑
        doMessage(message);
        //手工确认偏移量
        acknowledgment.acknowledge();
    }

    //处理消息逻辑
    private void doMessage(String message) {
        // 处理消息逻辑
        JsonNode jsonNode = JacksonUtil.parseTree(message);
        System.out.println("消息内容：" + jsonNode);
    }
}
